from kafka import KafkaConsumer
from kafka import KafkaProducer
import json
import os
import cv2
import numpy as np

from AlgorithmsLibrary.areaArrayAlgorithm.Algorithm_selection import select

from AlgorithmsLibrary.singleScanAlgorithm.singleScanSelect import singleScanprocess, CutsingleScan
from threading import Thread


def json_serializer(data):
    return json.dumps(data).encode("utf-8")


def consumerMessage(topicName):
    return KafkaConsumer(topicName, bootstrap_servers=['121.4.54.122:9092'],
                         auto_offset_reset='earliest',
                         group_id="consumer_group_" + topicName,
                         api_version=(0, 10, 1)
                         )




def singleScanlineImageAlgorithm(str_key, msg_value):
    """

    :param dict_key: 消费字典形式的key
    :param dict_value: 消费字典形式的value
    :return: 返回一个包含检测结果以及key的字典
    """
    img_byte = msg_value
    # 读取byte格式的图像
    img = cv2.imdecode(np.frombuffer(img_byte, np.uint8), cv2.IMREAD_UNCHANGED)

    print("处理线阵相机采集的车底大图")
    result = singleScanprocess(img)  # 处理线阵图像,得到诊断结果的列表
    # 解析字符串
    str_list = str_key.split('_')
    CutsingleScan(img, str_list[1])  # 将线扫的切割图片放入对应工单的子文件夹下,加入工单号
    str_check_return_list = []
    for index, name in enumerate(result):
        str_check_return_list.append("checkMsgKey"+"_"+str_list[1]+"_"+str_list[2]+"_"+str(index)+"_"+str_list[4]+"_"+str_list[5])


    str_value_check_return = result
    print("+" * 10)
    return {"key": str_check_return_list,
            "value": str_value_check_return}


def areaArrayImageAlgorithm(str_key, msg_value):
    """

    :param dict_key: 消费字符串形式的key
    :param dict_value: 消费字节码形式的value
    :return: 返回一个包含检测结果以及key的字典
    """
    img_byte = msg_value

    # 解析字符串
    str_list = str_key.split('_')
    # 面阵相机的数据落盘
    path = os.getcwd() + "/data/" + str_list[1]
    if not os.path.exists(path):
        os.makedirs(path)
    file_address = path + "/" +  str_list[2] + r".jpg"
    with open(file_address, 'wb') as f:
        f.write(img_byte)
    # 读取byte格式的图像
    img = cv2.imdecode(np.frombuffer(img_byte, np.uint8), cv2.IMREAD_UNCHANGED)
    # 选择项点对应的算法
    result = select(str_list[2] , img)
    print("正在处理面阵相机采集的图像")
    print("+" * 10)
    str_check_return = "checkMsgKey"+"_"+str_list[1]+"_"+str_list[2]+"_"+str_list[3]+"_"+str_list[4]+"_"+str_list[5]
    str_value_check_return = result
    return {"key": str_check_return,
            "value": str_value_check_return}


def array(producer):
    # 消费面阵相机图像topic信息
    consumer = consumerMessage("Itirs_img")
    print("starting the areaArray consumer")
    for msg in consumer:
        str_key = str(msg.key, encoding="utf-8")
        dict_check_return = areaArrayImageAlgorithm(str_key, msg.value)
        print("[%s] check result = {%s}"%(str_key.split('_')[2],dict_check_return["value"]))
        producer.send("Itirs_text", bytes(dict_check_return["value"], encoding = "utf8"),bytes(dict_check_return["key"], encoding = "utf8") )


def Line(producer):
    # 消费线阵相机图像topic信息
    consumer = consumerMessage("Itirs_finish_notice")
    print("starting the SingleScan consumer")
    for msg in consumer:
        str_key = str(msg.key, encoding="utf-8")
        dict_check_return = singleScanlineImageAlgorithm(str_key, msg.value)
        for i, key in enumerate(dict_check_return["key"]):
            print("[%s] check result = {%s}" % (str_key.split('_')[2]+str(i), dict_check_return["value"]['result'][i]))
            value = {"result": dict_check_return["value"]["result"][i]}
            producer.send("Itirs_text", bytes(value, encoding = "utf8"), bytes(key, encoding = "utf8"))

if __name__ == '__main__':
    # 实例化生产者对象,用于生产发送算法检测结果的消息
    producer = KafkaProducer(bootstrap_servers=['121.4.54.122:9092'],
                              api_version=(0, 10, 1)
                              )

    # 设置多线程进行面阵以及线阵图像资源的消费
    t1 = Thread(target=array, name="线程1",args=(producer,))
    t2 = Thread(target=Line, name="线程2",args=(producer,))
    t1.start()
    t2.start()

